package demo.netty.netty.websocket.time_server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.CharsetUtil;

import java.util.Date;
import java.util.Objects;

public class TimeWebSocketServer {

    public static void main(String[] args) {
        new TimeWebSocketServer().bind(8080);
    }

    public void bind(int port) {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap
                .group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                            //将请求和应答消息解码或者编码成http消息
                            .addLast("http-codec", new HttpServerCodec())
                            .addLast("aggregator", new HttpObjectAggregator(65536))
                            .addLast("http-chunked", new ChunkedWriteHandler())
                            .addLast("handler", new TimeWebSocketServerHandler());
                    }
                });
            Channel channelFuture = serverBootstrap.bind(port).sync().channel();
            System.out.println("Web Socket Server started at port :" + port + ".");
            System.out.println("open you browser and navigate to http://localhost:" + port + "/");
            channelFuture.closeFuture().sync();
        } catch (InterruptedException e) {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }

    }

    public static class TimeWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

        private WebSocketServerHandshaker webSocketServerHandshaker;

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
            if (msg instanceof FullHttpRequest) {
                this.handleHttpRequest(ctx, (FullHttpRequest) msg);
            } else {
                this.handleWebSocketRequest(ctx, (WebSocketFrame) msg);
            }
        }

        /**
         * 处理http请求
         * @param ctx
         * @param fullHttpRequest
         */
        private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest fullHttpRequest) {
            //http解码失败，并且协议并不是升级为websocket
            if (!fullHttpRequest.decoderResult().isSuccess() ||
                (!"websocket".equals(fullHttpRequest.headers().get("Upgrade")))) {
                    sendHttpResponse(ctx, fullHttpRequest, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1,
                                                                                       HttpResponseStatus.BAD_REQUEST));
                    return;
            }
            WebSocketServerHandshakerFactory webSocketServerHandshakerFactory
                = new WebSocketServerHandshakerFactory("ws://localhost:8080/websocket", null, false);
            webSocketServerHandshaker = webSocketServerHandshakerFactory.newHandshaker(fullHttpRequest);
            if (Objects.isNull(webSocketServerHandshaker)) {
                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
            } else {
                //创建握手时，就会将websocket的编码解码添加到管道中
                webSocketServerHandshaker.handshake(ctx.channel(), fullHttpRequest);
            }
        }

        private void handleWebSocketRequest(ChannelHandlerContext ctx, WebSocketFrame webSocketFrame) {
            //判断链路是否是关闭的指令
            if (webSocketFrame instanceof CloseWebSocketFrame) {
                webSocketServerHandshaker.close(ctx.channel(), (CloseWebSocketFrame) webSocketFrame.retain());
                return;
            }

            if (webSocketFrame instanceof PingWebSocketFrame) {
                ctx.channel().write(new PongWebSocketFrame(webSocketFrame.content().retain()));
                return;
            }

            //只支持文本消息，不支持二进制消息
            if (!(webSocketFrame instanceof TextWebSocketFrame)) {
                throw new UnsupportedOperationException(String.format("%s frame types not supported", webSocketFrame.getClass().getName()));
            }

            //返回应答消息
            String text = ((TextWebSocketFrame) webSocketFrame).text();
            ctx.channel().writeAndFlush(new TextWebSocketFrame(text + ", 欢迎使用netty websocket服务，现在是北京时间："+ new Date().toString()));
        }

        private static void sendHttpResponse(ChannelHandlerContext ctx,
                                             FullHttpRequest request,
                                             FullHttpResponse response) {
            if (response.status().code() != 200) {
                ByteBuf byteBuf = Unpooled.copiedBuffer(response.status().toString(), CharsetUtil.UTF_8);
                response.content().writeBytes(byteBuf);
                byteBuf.release();
                HttpUtil.setContentLength(response, response.content().readableBytes());
            }

            ChannelFuture future = ctx.channel().writeAndFlush(response);
            if (!HttpUtil.isKeepAlive(request) || response.status().code() != 200) {
                future.addListener(ChannelFutureListener.CLOSE);
            }
        }

    }


}
